package com.jg.visitor.kafka.listener.manually;

import com.jg.visitor.kafka.consumer.KafkaMsgHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;

@Slf4j
@Component
public class KafkaConsumerListener {

    @Resource
    private Map<String, KafkaMsgHandler> handlerMap;


    @KafkaListener(topics = {"#{'${spring.kafka.consume-topics}'.split(',')}"},
            groupId = "${spring.kafka.consumer.group-id}", containerFactory = "containerFactory")
    public void consumer(List<ConsumerRecord<String, String>> records) {
        for (ConsumerRecord<String, String> record : records) {
            try{
                String topic = record.topic();
                KafkaMsgHandler kafkaMsgHandler = handlerMap.get(topic);

                log.info("receive msg,topic={},handler is found? ==> {}",topic,kafkaMsgHandler != null);

                if(kafkaMsgHandler != null){
                    kafkaMsgHandler.onMsg(record.value());
                }
            }catch (Exception e){
                log.error("consume msg error",e);
            }
        }
    }
}
